package pub.cleangao.memm.transport.netty;


import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;
import pub.cleangao.memm.exception.ConnectionUnavailable;
import pub.cleangao.memm.exception.RpcException;
import pub.cleangao.memm.registry.RpcRuntime;
import pub.cleangao.memm.transport.TransportConst;
import pub.cleangao.memm.transport.message.MessageType;
import pub.cleangao.memm.transport.message.RpcMessage;
import pub.cleangao.memm.transport.message.RpcResponse;
import pub.cleangao.memm.transport.netty.compress.CompressType;
import pub.cleangao.memm.transport.netty.serialize.SerializationType;

import java.net.InetSocketAddress;
import java.net.SocketException;


@Slf4j
public class NettyRpcClientHandler extends ChannelInboundHandlerAdapter {

    private final UnprocessedRequests unprocessedRequests = UnprocessedRequests.getInstance();
    private static NettyRpcClient nettyRpcClient;


    /**
     * Read the message transmitted by the server
     */
    @Override
    @SuppressWarnings("unchecked")
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        try {
            if (msg instanceof RpcMessage tmp) {
                byte messageType = tmp.getMessageType();
                if (messageType == MessageType.HEARTBEAT_RESPONSE.value()) {
                    log.debug("Heartbeat [{}]", tmp.getData());
                } else if (messageType == MessageType.RESPONSE.value()) {
                    RpcResponse<Object> rpcResponse = (RpcResponse<Object>) tmp.getData();
                    unprocessedRequests.complete(rpcResponse);
                } else if (messageType == MessageType.EXCEPTION_RESPONSE.value()) {
                    RpcResponse<Object> rpcResponse = (RpcResponse<Object>) tmp.getData();
                    unprocessedRequests.complete(rpcResponse);
                    log.error("Rpc service method exception: {}", rpcResponse.getMessage());
                }
            }
        } finally {
            ReferenceCountUtil.release(msg);
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleState state = ((IdleStateEvent) evt).state();
            if (state == IdleState.WRITER_IDLE) {
                log.debug("Write idle happen [{}]", ctx.channel().remoteAddress());
                if (nettyRpcClient == null) {
                    nettyRpcClient = NettyRpcClient.getInstance();
                }
                Channel channel = nettyRpcClient.getChannel((InetSocketAddress) ctx.channel().remoteAddress());
                RpcMessage rpcMessage = new RpcMessage();
                rpcMessage.setCodec(SerializationType.PROTOSTUFF.getCode());
                rpcMessage.setCompress(CompressType.NONE.getCode());
                rpcMessage.setMessageType(MessageType.HEARTBEAT_REQUEST.value());
                rpcMessage.setData(TransportConst.PING);
                channel.writeAndFlush(rpcMessage).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    /**
     * Called when an exception occurs in processing a client message
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        if (cause instanceof ConnectionUnavailable unavailable) {
            RpcRuntime.addUnHealthServeNode(unavailable.getHost(), unavailable.getPort());
        } else if (cause instanceof SocketException) {
            log.info("Disconnected: " + ctx.name());
            RpcRuntime.downShelf();
        } else {
            log.error("Method invoke runtime exception: {}", cause.getMessage());
            cause.printStackTrace();
        }
        ctx.close();
    }
}
